package flink.eureka.foundation

import org.apache.flink.api.scala.ExecutionEnvironment
import scala.collection.mutable.ListBuffer

import org.apache.flink.api.scala._
/**
  *
  * @author com.ymy.hadoop   
  * @since 2019/6/4 0:14
  */
object JoinDemo01 {

  def main(args: Array[String]): Unit = {

    val env = ExecutionEnvironment.getExecutionEnvironment

    val info1 = ListBuffer[(Int, String)]()  // 编号  名字
    info1.append((1, "PK哥"));
    info1.append((2, "J哥"));
    info1.append((3, "小队长"));
    info1.append((4, "猪头呼"));

    val info2 = ListBuffer[(Int, String)]()  // 编号  城市
    info2.append((1, "北京"));
    info2.append((2, "上海"));
    info2.append((3, "成都"));
    info2.append((5, "杭州"));

    val data1 = env.fromCollection(info1)
    val data2 = env.fromCollection(info2)

    data1.join(data2).where(0).equalTo(0).apply((first,second)=>{
       if (second == null){
         (first._1,first._2,"-")
       }else{
         (first._1,first._2,second._2)
       }
    }).print()
  }
}
